package cn.zwx.remoting.netty;

import cn.zwx.config.FeignRpcProperties;
import cn.zwx.config.ServiceConfig;
import cn.zwx.constant.FeignRpcCommonConstant;
import cn.zwx.dispatcher.DispatcherServletProxy;
import cn.zwx.enumerate.HandleEnum;
import cn.zwx.mock.MockHttpServletRequest;
import cn.zwx.mock.MockHttpServletResponse;
import cn.zwx.remoting.RemotingCommand;
import cn.zwx.util.BeanFactoryUtils;
import com.alibaba.fastjson.JSON;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.tomcat.util.threads.TaskQueue;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.servlet.HandlerMapping;
import org.springframework.web.servlet.handler.HandlerMethodMappingNamingStrategy;
import org.springframework.web.servlet.mvc.method.RequestMappingInfo;
import org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerMapping;

import java.net.URI;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * NettyServer端
 * @author 105286
 */
public class NettyRemotingServer extends AbstractNettyRemoting implements RemotingServer {

    private static final Logger logger = LoggerFactory.getLogger(NettyRemotingServer.class);
    /**
     *  负责初始化netty服务器，并且开始监听端口的socket请求。
     **/
    private ServerBootstrap bootstrap;
    /**
     * 连接方式
     **/
    private volatile Channel serverChannel;
    /**
     *  工作线程组
     **/
    private NioEventLoopGroup workGroup;
    /**
     *  Boss线程组
     **/
    private NioEventLoopGroup bossGroup;
    /**
     *  工作线程下标
     **/
    private final AtomicInteger workThreadIndex = new AtomicInteger(0);
    /**
     *  线程池类
     **/
    private ThreadPoolExecutor poolExecutor;
    /**
     *  Boss线程下标
     **/
    private final AtomicInteger serverHandleThreadIndex = new AtomicInteger(0);
    /**
     *  执行定时任务线程池
     **/
    ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1,new BasicThreadFactory.Builder()
                .namingPattern(FeignRpcCommonConstant.SCHEDULED_EXECUTOR_THREAD_NAME_SUFFIX)
                .daemon(Boolean.TRUE)
                .priority( Thread.NORM_PRIORITY )
                .build());

    private final FeignRpcProperties feignRpcProperties;

    private final DispatcherServletProxy dispatcherServlet;

    public NettyRemotingServer(FeignRpcProperties feignRpcProperties, DispatcherServletProxy dispatcherServlet) {
        this.feignRpcProperties = feignRpcProperties;
        this.dispatcherServlet = dispatcherServlet;
    }

    @Override
    public void start() {
        ServiceConfig service = feignRpcProperties.getService();
        // 构建阻塞队列
        TaskQueue taskQueue = new TaskQueue(service.getMaxTaskQueueSize());
        // 构建请求线程池
        poolExecutor = new ThreadPoolExecutor(service.getCoreThreadNum(), service.getMaxThreadNum(),  service.getMaxTheadFreeTime(), TimeUnit.SECONDS,taskQueue,
                r -> new Thread(r, FeignRpcCommonConstant.RPC_SERVER_HANDLE_THREAD + serverHandleThreadIndex.getAndIncrement()));
        taskQueue.setParent(poolExecutor);
        // 构建工作线程池
        workGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2, r -> {
            return new Thread(r, FeignRpcCommonConstant.NETTY_SERVER_WORK_THREAD + workThreadIndex.getAndIncrement());
        });
        // 构建boss线程
        bossGroup = new NioEventLoopGroup(1, r -> {
            return new Thread(r, FeignRpcCommonConstant.NETTY_SERVER_ACCEPT_THREAD);
        });
        bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup,workGroup)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_RCVBUF, 1024 * 100*10)
                .childOption(ChannelOption.SO_SNDBUF, 1024 * 100*10)
                .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 0, 8, 0, 8));
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new RemotingServerHandler());
                        ch.pipeline().addLast(new RemotingCommandHandle());
                        ch.pipeline().addFirst(new StringEncoder());
                        ch.pipeline().addFirst(new LengthFieldPrepender(8));
                    }
                });
        // init scheduleAtFixedRate
        scheduledExecutorService.scheduleAtFixedRate(()->{
            try {
                //扫描结果响应表
                scanResponseTable();
            } catch (Exception e) {
                logger.error("scanResponseTable exception", e);
            }
        },1000 * 3L, 1000 * 10L,TimeUnit.MILLISECONDS);
    }


    /**
     * 构建好HttpServletRequest和HttpServletResponse对象，并调用DispatcherServlet上面的service方法
     * <code>
     *     因为我们的SpringMVC 底层是Http协议的 所以我们在拿到请求参数和返回参数
     *     需要进行转换为HttpRequest和HttpResponse 不然这块存在问题
     * </code>
     * <待优化>
     *    这里其实有更好的方式，我们可以构建我们自己的HttpResponse和HttpRequest
     *    最终把数据写出去 或者我们这里使用GRPC去发起调用
     * </优化>
     * @param ctx ChannelHandlerContext
     * @param cmd RemotingCommand
     */
    @Override
    protected void processRequestCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        try {
            poolExecutor.execute(() -> handleMessage(ctx, cmd));
        } catch (RejectedExecutionException e) {
            logger.error("to many remoting connection");
            RemotingCommand rp = new RemotingCommand();
            rp.setType(HandleEnum.RESPONSE_COMMAND.getCode());
            rp.setXid(cmd.getXid());
            rp.setCode(500);
            rp.setError("to many remoting connection ,reject remoting connect");
            ctx.pipeline().writeAndFlush(rp);
        } catch (Throwable throwable) {
            logger.error(throwable.toString());
            RemotingCommand rp = new RemotingCommand();
            rp.setType(HandleEnum.RESPONSE_COMMAND.getCode());
            rp.setXid(cmd.getXid());
            rp.setCode(500);
            rp.setError(throwable.toString());
            ctx.pipeline().writeAndFlush(rp);
        }

    }

    private void handleMessage(ChannelHandlerContext ctx, RemotingCommand cmd) {
        MockHttpServletRequest request = new MockHttpServletRequest();
        MockHttpServletResponse response = new MockHttpServletResponse();
        String xid = cmd.getXid();
        RemotingCommand rp = new RemotingCommand();
        rp.setXid(xid);
        rp.setType(HandleEnum.RESPONSE_COMMAND.getCode());
        this.transformRemotingCommandToRequest(cmd, request);
        try {
            dispatcherServlet.doProcessRequest(request, response);
            this.transformResponseToRemotingCommand(response, rp);
        } catch (Throwable e) {
            logger.error(e.toString());
            this.transformThrowableToRemotingCommand(response, rp, 500, e.toString());
        }
        // 写出结果
        ctx.pipeline().writeAndFlush(rp);

    }

    /**
     * 处理异常情况
     * @param response 返回番薯
     * @param rp RemotingCommand
     * @param errorCode 错误码
     * @param errorMsg 错误信息
     */
    private void transformThrowableToRemotingCommand(MockHttpServletResponse response, RemotingCommand rp, int errorCode, String errorMsg) {
        rp.setCode(errorCode);
        rp.setError(errorMsg);
        addHeaderToRemotingCommand(response, rp);
    }

    /**
     * 处理返回参数
     * @param response  response
     * @param rp rp
     */
    private void transformResponseToRemotingCommand(MockHttpServletResponse response, RemotingCommand rp) {
        rp.setBody(response.getContentAsByteArray());
        rp.setCode(response.getStatus());
        this.addHeaderToRemotingCommand(response, rp);
    }

    /**
     * 处理请求参数组装
     * @param cmd RemotingCommand
     * @param mockRequest MockHttpServletRequest
     */
    private void transformRemotingCommandToRequest(RemotingCommand cmd, MockHttpServletRequest mockRequest) {
        recodeHeadersToRequest(cmd, mockRequest);
        URI uri = URI.create(cmd.getUrl());
        mockRequest.setContent(cmd.getBody());
        mockRequest.setRequestUri(uri.getPath());
        mockRequest.setServletPath(uri.getPath());
        mockRequest.setQueryString(uri.getQuery());
        // 如果是GET请求提取请求参数
        if (!StringUtils.isBlank(uri.getQuery())) {
            String[] parameters = uri.getQuery().split("&");
            for (String parameter : parameters) {
                if (StringUtils.isBlank(parameter)) {
                    continue;
                }
                String[] param = parameter.split("=");
                String key = param[0];
                String value = "";
                if (param.length >= 2) {
                    StringBuilder builder = new StringBuilder();
                    for (int i = 1; i < param.length; i++) {
                        builder.append(param[i].trim()).append(",");
                    }
                    value = builder.substring(0, builder.lastIndexOf(","));
                }
                mockRequest.setParameter(key, value);
            }
        }
        mockRequest.setPathInfo(uri.getQuery());
        mockRequest.setMethod(cmd.getMethod());
    }


    /**
     * 将RemotingCommand中header成员变量设置到HttpServletRequest的请求头中去
     * @param rq rq
     * @param request MockHttpServletRequest
     */
    private void recodeHeadersToRequest(RemotingCommand rq, MockHttpServletRequest request) {
        Map<String, Collection<String>> headers = rq.getHeader();
        if (null != headers) {
            for (String key : headers.keySet()) {
                Collection<String> values = headers.get(key);
                if (values != null) {
                    for (Object value : values) {
                        request.addHeader(key, value);
                    }
                }
            }
        }
    }


    /**
     * 将HttpServletResponse中的请求头添加到RemotingCommand对象中
     * @param response  response
     * @param rp RemotingCommand
     */
    private void addHeaderToRemotingCommand(MockHttpServletResponse response, RemotingCommand rp) {
        Collection<String> headerNames = response.getHeaderNames();
        if (null != headerNames) {
            Map<String, Collection<String>> headers = new HashMap<>(8);
            for (String headerName : headerNames) {
                headers.put(headerName, response.getHeaders(headerName));
            }
            rp.setHeader(headers);
        }
    }


    /**
     * 端口绑定
     * @param port 端口
     * @return void
     * @author zhangwenxue
     * @createTime 2022/6/20 14:15
     **/
    @Override
    public void bind(int port) {
         try {
             bootstrap.bind(port).sync().addListener((ChannelFutureListener) future -> {
                 if (future.isSuccess()) {
                     serverChannel = future.channel();
                     if (logger.isInfoEnabled()){
                         logger.info("netty server bind port:{} success", port);
                     }
                     return;
                 }
                 if (logger.isWarnEnabled()){
                     logger.warn("netty server in port:{} fail,cause:{}", port, future.cause().getMessage());
                 }
                 serverChannel = null;
             });
         } catch (InterruptedException e) {
             if (logger.isWarnEnabled()){
                 logger.warn("netty server start fail,bind {} fail,cause:{}", port, e);
             }
         }
    }

    @Override
    public boolean isActive() {
        return (serverChannel == null || !serverChannel.isActive()) ? Boolean.FALSE:Boolean.TRUE;
    }



    @Override
    public void shutdown() {
        // 取消定时任务
        scheduledExecutorService.shutdown();
        // 关闭工作线程
        if (workGroup != null) {
            workGroup.shutdownGracefully();
        }
        // 关闭Boss线程
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }
        if (poolExecutor != null) {
            poolExecutor.shutdown();
        }
    }


    private class RemotingServerHandler extends SimpleChannelInboundHandler<String> {

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            if (logger.isInfoEnabled()){
                logger.info("remoting client connected current server success,remoting address {}",ctx.channel().remoteAddress());
            }
            super.channelActive(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            if (logger.isWarnEnabled()){
                logger.warn("{} broken connect,cause:{}", ctx.channel().remoteAddress(), cause.toString());
            }
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            RemotingCommand remotingCommand = JSON.parseObject(msg, RemotingCommand.class);
            processMessageReceived(ctx, remotingCommand);
        }
    }
}
